package pl.softech.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.logging.Level;
import java.util.logging.Logger;

import pl.softech.logger.MyLogger;
import pl.softech.server.http.HttpBadRequestException;
import pl.softech.server.http.HttpRequestProcessor;

public class Server implements Runnable {

    protected Logger logger = MyLogger.getLogger(getClass());
    
    private static final int DEFAULT_WORKER_POOL_SIZE = 100;
    private static final int DEFAULT_BACKLOG_SIZE = 100;
    private static final int DEFAULT_READ_BUFFER_SIZE = 8 * 1024;

    private final int port;

    private final int backlog;

    // kanal na ktorym akceptujemy polaczenia
    private ServerSocketChannel serverChannel;
    // selektor ktory bedzie sprawdzal gotowosc kanallow do operacji io
    private Selector selector;

    final List<SocketCommand> socketCommands;

    private final WorkerManager workerManager;

    private final IRequestProcessor requestProcessor;

    private ByteBuffer readBuffer;

    public Server(int port, int backlog, int poolSize, IRequestProcessor requestProcessor) throws IOException {
        this.port = port;
        this.backlog = backlog;
        this.requestProcessor = requestProcessor;
        this.workerManager = new WorkerManager(poolSize, this);
        this.socketCommands = new LinkedList<SocketCommand>();
        init();
    }

    public Server(int port, IRequestProcessor requestProcessor) throws IOException {
        this(port, DEFAULT_BACKLOG_SIZE, DEFAULT_WORKER_POOL_SIZE, requestProcessor);
    }

    private void init() throws IOException {
        logger.info("Initialising");
        logger.info("Server listen port: " + port);
        logger.info("Server backlog size: " + backlog);
        
        serverChannel = ServerSocketChannel.open();
        // Chcemy aby kanal byl nieblokujacy
        serverChannel.configureBlocking(false);

        // Bindujemy kanal do danego portu
        InetSocketAddress isa = new InetSocketAddress(port);
        serverChannel.socket().bind(isa, backlog);

        // Tworzymy selektor
        selector = SelectorProvider.provider().openSelector();
        // rejestrujemy selektor aby reagowal na proby nawiazania polaczenia
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);

        readBuffer = ByteBuffer.allocate(DEFAULT_READ_BUFFER_SIZE);
    }

    private void processSocketCommands() {
        synchronized (socketCommands) {
            for (SocketCommand command : socketCommands) {
                switch (command.type) {
                case SocketCommand.CHANGEOPS:
                    SelectionKey key = command.socket.keyFor(selector);
                    key.interestOps(command.ops);
                    break;
                }
            }
            socketCommands.clear();
        }
    }

    @Override
    public void run() {
        logger.info("Starting listening on port: " + port);
        while (true) {
            try {

                processSocketCommands();

                // czekamy na zmiane gotowosci kanalow
                selector.select();

                Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                while (it.hasNext()) {
                    SelectionKey key = it.next();
                    it.remove();

                    if (!key.isValid()) {
                        continue;
                    }

                    // sprawdzamy czy kanal jest gotowy do jego obslugi
                    if (key.isAcceptable()) {
                        accept(key);
                    } else if (key.isReadable()) {
                        read(key);
                    } else if (key.isWritable()) {
                        write(key);
                    }

                }

            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }

    private void accept(SelectionKey key) throws IOException {
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
        SocketChannel socketChannel = serverSocketChannel.accept();
        // Socket socket = socketChannel.socket();
        socketChannel.configureBlocking(false);
        // rejestrujemy gotowosc soketu od czytania
        socketChannel.register(this.selector, SelectionKey.OP_READ);
    }

    private void read(SelectionKey key) throws IOException, HttpBadRequestException {
        SocketChannel socketChannel = (SocketChannel) key.channel();
        readBuffer.clear();
        int count;
        try {
            count = socketChannel.read(readBuffer);
        } catch (Exception e) {
            logger.log(Level.WARNING, "Client has been dissconnected", e);
            // klient mogl sie po prostu rozlaczyc niespodziewanie
            key.cancel();
            socketChannel.close();
            return;
        }

        if (count == -1) {
            // klient zamknal polaczenie
            logger.log(Level.FINE, "Client closed connection");
            key.channel().close();
            key.cancel();
            return;
        }

        workerManager.processData(socketChannel, this.readBuffer.array(), count);
    }

    private void write(SelectionKey key) throws IOException {
        SocketChannel socketChannel = (SocketChannel) key.channel();
        synchronized (workerManager.getPendingData()) {
            Queue<IResponse> queue = workerManager.getPendingData().get(socketChannel);
            IResponse response = null;
            while (!queue.isEmpty()) {
                response = queue.peek();
                ByteBuffer buf = (ByteBuffer) ByteBuffer.wrap(response.getRawData());
                socketChannel.write(buf);
                if (buf.remaining() > 0) {
                    break;
                }
                queue.poll();
            }

            if (queue.isEmpty()) {
                // aby zakonczyc polaczenie
                if (response != null && response.closeConnection()) {
                    key.channel().close();
                    key.cancel();
                } else {
                    //jesli nie chcemy konczyc no to daleje sluchay
                    key.interestOps(SelectionKey.OP_READ);
                }
            }

        }
    }

    public void wakeup() {
        selector.wakeup();
    }

    public IRequestProcessor getRequestProcessor() {
        return requestProcessor;
    }

    public static void main(String[] args) throws IOException {
        Server httpServer = new Server(8080, new HttpRequestProcessor());
        httpServer.run();
    }

}
